/**
 * Copyright 2014 opencxa.org
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.cxa.rpc.cloud.server;

import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadPoolExecutor;

import org.cxa.exec.simpleThread.SimpleThread;
import org.cxa.rpc.cloud.server.components.RootComponent;
import org.cxa.rpc.cloud.server.components.authentication.DatabaseManager;
import org.cxa.rpc.cloud.server.components.authentication.Group;
import org.cxa.rpc.cloud.server.components.authentication.User;
import org.cxa.rpc.messages.NotificationRpcMessage;
import org.cxa.rpc.messages.RequestRpcMessage;
import org.cxa.rpc.messages.ResponseRpcMessage;
import org.cxa.rpc.messages.RpcMessage;


/**
 * @author Christopher Armenio
 */
public class IoBroker extends SimpleThread
{
	private static final int NUM_UPDATE_THREADS = 5;
	private static IoBroker SINGLETON = null;
	
	
	private class ClientUpdater implements Runnable
	{
		private final DeviceProxy dp;
		private ClientUpdater(DeviceProxy dpIn) { this.dp = dpIn; }
		
		@Override
		public void run() { this.dp.update(); }
	}
	
	private final ThreadPoolExecutor threadPool_clientUpdater = (ThreadPoolExecutor)Executors.newFixedThreadPool(NUM_UPDATE_THREADS);
	private final RootComponent rc;
	
	private final Map<String, DeviceProxy> devices = Collections.synchronizedMap(new HashMap<String, DeviceProxy>());
	
	
	private IoBroker()
	{
		this.rc = new RootComponent();
		this.threadPool_clientUpdater.setRejectedExecutionHandler(new ThreadPoolExecutor.CallerRunsPolicy());
	}
	
	
	public void addDeviceProxy(DeviceProxy dpIn)
	{
		if( dpIn == null ) throw new IllegalArgumentException("deviceProxy must NOT be null");
		this.devices.put(dpIn.getSerialNumber(), dpIn);
		this.logger.info("new deviceProxy " + dpIn.toString());
	}
	
	
	public void removeDisconnectedDeviceProxy(DeviceProxy dpIn)
	{
		this.devices.remove(dpIn);
		this.logger.info("removed deviceProxy " + dpIn.toString());
	}
	
	
	public DeviceProxy getDeviceProxy(String serialNumIn)
	{
		return this.devices.get(serialNumIn);
	}
	
	
	public void routeMessage(RpcMessage msgIn)
	{
		this.logger.trace(String.format("routing message %s", msgIn.toString()));
		
		switch( msgIn.getMsgType() )
		{
			case REQUEST:
				// for now, we'll assume this message is bound for a system component / method
				// if it's bound for a DeviceProxy, the permissions will be updated in DeviceProxy
				DeviceProxy sourceDp = DeviceProxy.extractSourceDeviceProxy_fromMessageMetaData(msgIn);
				List<Group> sourceDpGroups = sourceDp.getGroups();
				User sourceDpOwner = sourceDp.getOwner();
				msgIn.setPermissions((sourceDpGroups != null) && (sourceDpGroups.contains(DatabaseManager.GROUP_ADMIN)), (sourceDpOwner != null) && (sourceDpOwner.equals(DatabaseManager.USER_ROOT)));
				
				this.rc.onRequestReceived((RequestRpcMessage)msgIn);
				break;
				
			case RESPONSE:
				this.rc.onUnmatchedResponseReceived((ResponseRpcMessage)msgIn);
				break;
			
			case NOTIFICATION:
				this.rc.onNotificationReceived((NotificationRpcMessage)msgIn);
				break;
				
			default:
				// do nothing
				break;
		}
	}
	
	
	public Collection<DeviceProxy> getConnectedClients()
	{
		return this.devices.values();
	}


	@Override
	public void run()
	{
		while( this.shouldKeepRunning() )
		{
			synchronized(this.devices)
			{
				for( DeviceProxy currDp : this.devices.values() )
				{
					if( !currDp.isUpdating() ) this.threadPool_clientUpdater.execute(new ClientUpdater(currDp));
				}
			}
			try{ Thread.sleep(100); } catch(InterruptedException e) { }
		}
	}
	
	
	public static boolean init()
	{
		if( SINGLETON != null ) throw new RuntimeException("multiple calls to .init()");
		SINGLETON = new IoBroker();
		return SINGLETON.start();
	}
	
	
	public static IoBroker getSingleton()
	{
		if( SINGLETON == null ) throw new RuntimeException(".init() must be called first");
		return SINGLETON;
	}
}
